package com.atguigu.gmall;

/**
 * @ClassName: CanalClientAnalyze
 * @Description:
 * @Author: kele
 * @Date: 2021/3/26 9:29
 **/

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.atguigu.gmall.utils.Constant.KafkaConstant;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * 灵活分析需求
 *  1、配置连接
 *  2、开启连接
 *  3、订阅相关的表
 *  4、获取数据
 *      order_info(insert)、
 *      order_detail(insert)、
 *      userinfo(insert、update)
 *  5、处理数据、
 *  6、将数据写入kafka中、
 *
 *
 *
 */
public class CanalClientAnalyze {


    public static void main(String[] args) throws InvalidProtocolBufferException, InterruptedException {

//        1、获取连接
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("hadoop102", 11111),
                "example",
                "",
                ""
        );

        canalConnector.connect();
        canalConnector.subscribe("d1026.*");

        while (true){

            Message message = canalConnector.get(100);

            if(message.getId() == -1){
                Thread.sleep(1000);
                continue;
            }


            List<CanalEntry.Entry> entries = message.getEntries();


            for (CanalEntry.Entry entry : entries) {

                if (entry.getEntryType().equals(CanalEntry.EntryType.ROWDATA)) {

                    ByteString storeValue = entry.getStoreValue();

                    //得到序列化的结果
                    CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);

                    //保存order_info的数据
                    if(entry.getHeader().getTableName().equals("order_info")
                            && rowChange.getEventType().equals(CanalEntry.EventType.INSERT) ){

                        handler(rowChange,KafkaConstant.KAFKA_TOPIC_ORDER_INFO);
                    }

                    if(entry.getHeader().getTableName().equals("order_detail")
                            && rowChange.getEventType().equals(CanalEntry.EventType.INSERT) ){

                        handler(rowChange,KafkaConstant.KAFKA_TOPIC_ORDER_DETAIL);
                    }

                    if(entry.getHeader().getTableName().equals("user_info")
                            && ( rowChange.getEventType().equals(CanalEntry.EventType.INSERT)
                            || rowChange.getEventType().equals(CanalEntry.EventType.UPDATE) )
                    ){

                        handler(rowChange,KafkaConstant.KAFKA_TOPIC_USER_INFO);
                    }
                }
            }

        }
    }



    public static void handler(CanalEntry.RowChange rowChange,String topicName){


        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();


        for (CanalEntry.RowData rowData : rowDatasList) {


            List<CanalEntry.Column> columnsList = rowData.getAfterColumnsList();

            JSONObject jsonObject = new JSONObject();

            for (CanalEntry.Column column : columnsList) {

                jsonObject.put(column.getName(), column.getValue());

            }

            MyProducer.send(topicName,jsonObject.toString());

        }

    }

}